package customized_input;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.util.LineReader;

import java.io.IOException;
import java.util.List;

public class MyInputFormat extends FileInputFormat<LongWritable, Text> {

    @Override
    public RecordReader<LongWritable, Text> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        return new MyRecordReader();
    }

    @Override
    protected boolean isSplitable(JobContext context, Path filename){
        return false;
    }

    @Override
    public List<InputSplit> getSplits(JobContext job) throws IOException {
        return super.getSplits(job);
    }

    public static class MyRecordReader extends RecordReader<LongWritable, Text> {

        private long start;
        private long end;
        private long pos;

        private FSDataInputStream fin = null;
        // define key of each k-v pair
        private LongWritable key = null;
        // define value of each k-v pair
        private Text value = null;
        // define LineReader for FSDataInputStream
        private LineReader reader = null;

        @Override
        public void initialize(InputSplit split, TaskAttemptContext context)
                throws IOException, InterruptedException {
            FileSplit fileSplit = (FileSplit) split;
            start = fileSplit.getStart();
            end = start + fileSplit.getLength();
            Configuration conf = context.getConfiguration();
            Path path = fileSplit.getPath();
            // prepare a line reader for reading
            FileSystem fs = path.getFileSystem(conf);
            FSDataInputStream fin = fs.open(path);
            fin.seek(start);
            reader = new LineReader(fin);
            pos = 1;
        }

        @Override
        public boolean nextKeyValue()
                throws IOException, InterruptedException {
            if (key == null) {
                key = new LongWritable();
            }
            key.set(pos);
            if (value == null) {
                value = new Text();
            }
            if (reader.readLine(value) == 0) {
                return false;
            }
            pos++;
            return true;
        }

        @Override
        public LongWritable getCurrentKey() throws IOException, InterruptedException {
            return key;
        }

        @Override
        public Text getCurrentValue() throws IOException, InterruptedException {
            return value;
        }

        @Override
        public float getProgress() throws IOException, InterruptedException {
            return 0;
        }

        @Override
        public void close() throws IOException {
            if (fin != null) fin.close();
        }
    }
}
